package com.xf.zk.curator.example;

import com.xf.zk.curator.constant.ZkConstant;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.atomic.AtomicValue;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicInteger;
import org.apache.curator.framework.recipes.queue.DistributedDelayQueue;
import org.apache.curator.framework.recipes.queue.DistributedIdQueue;
import org.apache.curator.framework.recipes.queue.DistributedPriorityQueue;
import org.apache.curator.framework.recipes.queue.DistributedQueue;
import org.apache.curator.framework.recipes.queue.QueueBuilder;
import org.apache.curator.framework.recipes.queue.QueueConsumer;
import org.apache.curator.framework.recipes.queue.QueueSerializer;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.retry.RetryNTimes;

public class DistributedQueueTest {
	static String distributePath= "/curator_recipes_distributedQueue";
	static CuratorFramework client = CuratorFrameworkFactory.builder().connectString(ZkConstant.SERVER)
			.retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();

	public static void main(String[] args) throws Exception {
		client.start();
		QueueConsumer<String> consumer = new QueueConsumer<String>() {
			@Override
			public void consumeMessage(String message) throws Exception {
				System.out.println(message);
			}
			@Override
			public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
			}
		};
		QueueSerializer<String> serializer = new QueueSerializer<String>() {
			@Override
			public byte[] serialize(String item) {
				return item.getBytes();
			}
			@Override
			public String deserialize(byte[] bytes) {
				return new String(bytes);
			}
		};
		QueueBuilder<String> builder = QueueBuilder.builder(client, consumer, serializer, distributePath);
//		DistributedDelayQueue<String> delayQueue = builder.buildDelayQueue();
//		DistributedIdQueue<String> idQueue = builder.buildIdQueue();
//		DistributedPriorityQueue<String> priorityQueue = builder.buildPriorityQueue(1);
		final DistributedQueue<String> queue = builder.buildQueue();
		// Start the queue. No other methods work until this is called
		queue.start();
		for (int i = 0; i < 10; i++) {
			Thread thread = new Thread(new Runnable() {
				@Override
				public void run() {
					try {
						queue.put(Thread.currentThread().getName());
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			});
			thread.start();
		}
		Thread.sleep(Integer.MAX_VALUE);
	}
}
